#!/opt/anaconda2/bin/python2.7
# -*- coding: utf8 -*-

'''
Created on 2018年10月11日

@author: wzt
设计思路:
1：datax脚本要使用命令行模式调用，采用线程池模式
2：在线程里读取datax结束信号，如果发生有一个出错的情况，py进程返回-1，表示错误，由smartCantor来决定重跑
3：json文件存放在job目录中，每一个job文件就是一个组任务
7：关键问题：并发控制(归结为py的多线程控制)，这次先不控制，采用顺序调用的方法

'''

import sys
import os
import signal
import subprocess
import json
import MySQLdb
import time
import datetime
import subprocess
import random
from collections import defaultdict, OrderedDict
from clickhouse_driver.client import Client
import logging
#from concurrent.futures import ThreadPoolExecutor
#from concurrent.futures import as_completed
from multiprocessing import Pool
import os, time



gUrls=[]
gUserName=''
gPasswd=''
NAME='NAME'
UPDATE='UPDATE'
SELECT='SELECT'
gDataBase=''
gPorts=[]

# logging.basicConfig函数对日志的输出格
logging.basicConfig(encode="utf-8" , encoding = "UTF-8",level=logging.DEBUG,
        format='%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s')

#扫描目录,获取文件列表
def scanDir(directory,prefix="job-",postfix=".json"):
    filesList=[]

    for root, sub_dirs, files in os.walk(directory):
        for special_file in files:
            if postfix:
                if special_file.endswith(postfix):
                    filesList.append(os.path.join(root,special_file))
            elif prefix:
                if special_file.startswith(prefix):
                    filesList.append(os.path.join(root,special_file))
            else:
                filesList.append(os.path.join(root,special_file))

    print now(),"目录下有文件",directory,filesList
    return filesList

#读取文件列表，并转为json
def readFiles(fileList):
    dictJson = {}
    dictJson = OrderedDict()
    for filePath in fileList:
        try:
            strContent = ''
            f = open(filePath, 'r')
            #try:
            #   print json.load(f)
            #except BaseException:
            #   print "error"

            lineList = f.readlines()
            for i in range(0, len(lineList)):
                lineList[i] = lineList[i].strip('\n')
                strContent += lineList[i]


            #str转json对象
            #object_pairs_hook=OrderedDict 让json转化时按照文本顺序
            dictJson[lastPosString(filePath)] = json.loads(strContent,object_pairs_hook=OrderedDict)
            #print(dictJson)

        finally:
            if f:
               f.close()
    return dictJson



#字符串最后出现的位置
def lastPosString(str,strFind="/"):
    pos = str.rfind("/")
    return str[pos+1:]

#获取昨天的日期
def getYesterday():
    today=datetime.date.today()
    oneday=datetime.timedelta(days=1)
    yesterday=today-oneday
    #time.strftime("%Y%m%d%H%M%S",yesterday)
    return yesterday.strftime('%Y%m%d')

#获取今天的日期
def getToday():
    today=datetime.date.today()
    return today.strftime('%Y%m%d')

def now():
    return time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))


def getIpUserPasswd(dictJson):
    global gUrls
    gUrls = dictJson['ckjdbcurls']['urls']
    global gUserName
    gUserName = dictJson['ckjdbcurls']['user']
    global gPasswd
    gPasswd = dictJson['ckjdbcurls']['password']
    global gPorts
    gPorts = dictJson['ckjdbcurls']['ports']
    global gDataBase
    gDataBase = dictJson['ckjdbcurls']['database']

    print now(),'Urls:',gUrls,' user:',gUserName,' ports:',gPorts,' database:',gDataBase


#因为需要固定线程的数量，以避免所有的组都启动对ck db造成高的负载...
def getSQLAndExcute(dictJson):
    #print now(),'Parent process %s.' % os.getpid()
    logging.info("父进程 %s",os.getpid())
    try:
        pool = Pool()
        i=0
        for mapSQL in dictJson['intf2model']:
            i+=1
            encode_SQL = json.dumps(mapSQL)
            #print now(),"执行第 ",i,"组SQL",encode_SQL
            logging.info("第 %d 组SQL:%s",i,encode_SQL)
            pool.apply_async(executeSQLFunc, args=(encode_SQL,))
            #executeSQLFunc(encode_SQL)

        logging.info("Waiting for all subprocesses done...")
        pool.close()
        pool.join()
        logging.info("All subprocesses done.")
    except Exception as e:
        print e
        sys.exit(-1)

def testfun1(a):
    for i in a:
        print i

def executeSQLFunc(encode_SQL):
        logging.debug("子进程 %s 收到编码后的SQL %s",os.getpid(),encode_SQL)
        mapSQL = json.loads(encode_SQL,object_pairs_hook=OrderedDict)
        sqlList = []
        for (key,sql) in mapSQL.items():
            #组内的SQL顺序执行
            #print now(),"key:",key
            if UPDATE == key :
                #print now(),'执行DML SQL drop,delete,每个连接都要执行,以覆盖所有的ck master节点'
                logging.info("执行DML SQL drop,delete,每个连接都要执行,以覆盖所有的ck master节点")
                executeUpdateSQLOnCk(sql)
            elif  SELECT in key :
                #在这里先将SQL语句保存到list里,等这个循环执行完成以后，在使用循环单独执行list中的SQL
                sqlList.append(sql)
            elif  NAME in key :
                print now(),"模型名称:",sql
                #encode 方便汉字的用logging 记录，这里缘由复杂，从字符串到JSON，JSON在反序列化为字典，需要以unicode编码，
                # 而loggin 不能直接接收unicode字符串，所以需要编码为utf8
                logging.info("模型名称 %s",sql.encode('utf-8'))
            else:
                print now(),"不支持的语句",sql
        #执行QUERY类型的SQL，顺序执行
        executeQuerySQLOnCk(sqlList)
        return 0

#connection clickhouse
def executeUpdateSQLOnCk(updateSQL):
    #for 循环的方式执行urls中所有的ip
    try:
        for i in range(0,len(gUrls)):
            #print now(),"SQL运行环境",gUrls[i],":",gPorts[i],"执行SQL:",updateSQL
            logging.info("SQL运行环境 urls:%s, ports:%s, 执行SQL %s",str(gUrls[i]),str(gPorts[i]),str(updateSQL))
            client = Client(host=gUrls[i],port=gPorts[i],database=gDataBase,user=gUserName,password=gPasswd)
            client.execute(updateSQL)
            client.disconnect()
    except Exception as e:
        print now(),e
        sys.exit(-1)



def executeQuerySQLOnCk(querySQLList):
    #随机选一个能执行的ip，在执行之前要先测试连接
    try:
        size1 = len(gUrls)-1 if len(gUrls)> 1 else 0
        rndIndex = random.randint(0,size1)
        #print now(),"选择环境 长度=",size1,"随机值=",rndIndex,",",gUrls[rndIndex],":",gPorts[rndIndex],"执行SQL组:",querySQLList
        logging.info("选择环境 urls配置了 %d 个,随机值:%d,%s:%s,执行SQL组:%s",size1,rndIndex,str(gUrls[rndIndex]),str(gPorts[rndIndex]),str(querySQLList))
        client = Client(host=gUrls[rndIndex],port=gPorts[rndIndex],database=gDataBase,user=gUserName,password=gPasswd)
        for querySQL in querySQLList:
            client.execute(querySQL)
            #print now(),"执行SQL：",querySQL
            logging.info("现在执行SQL：%s ",str(querySQL))
        client.disconnect()
    except Exception as e:
        print now(),e
        sys.exit(-1)



if __name__ == "__main__":
    #扫描目录，读取文件,转为json object
    #执行json中SQL语句{...} 里面顺序执行，{},{},... 之间并发执行，使用多进程同时执行(多线程参数传递有问题)
    #anaconda2 2.7 环境 /opt/anaconda2
    sys.argv[1]= '/workdir/git_repository/python/clickhousesqldo/job_intf2model'
    #sys.argv[1]= '/workdir/git_repository/python/clickhousesqldo/job_test'
    dirDataWareHouse = sys.argv[1]
    #print now(),'输入目录：',sys.argv[1]
    logging.info("输入目录：%s ",sys.argv[1])
    fileList = scanDir(dirDataWareHouse)
    if len(fileList) > 0 :
        dictJson = readFiles(fileList)
    else:
        #print now(),dirDataWareHouse,'目录下没有文件'
        logging.info(" %s 目录下没有文件",dirDataWareHouse)
        sys.exit(0)

    #dictJson包含了目录下所有的json文件
    for keyItem in dictJson:
        #print now(),'文件名称:',keyItem
        logging.info("文件名称:%s",keyItem)
        getIpUserPasswd(dictJson[keyItem])
        getSQLAndExcute(dictJson[keyItem])


